package day05;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import untils.ClickEvent;
import untils.ClickSource;
import java.sql.Timestamp;

/**
 * 触发器
 * 窗口的第一条数据到达以后，第一条数据的时间戳后面的整数秒，触发窗口计算
 */
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ClickEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<ClickEvent>() {
                                    @Override
                                    public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                }
                        )
                )
                .keyBy(r -> r.url)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                //设置触发器
                .trigger(new MyTrigger())
                .process(new ProcessWindowFunction<ClickEvent, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<ClickEvent> elements, Collector<String> out) throws Exception {
                        out.collect("url是：" + s + "的窗口" + new Timestamp(context.window().getStart()) + "~" +
                                "" + new Timestamp(context.window().getEnd()) + "的pv次数是：" + elements.spliterator().getExactSizeIfKnown());
                    }
                })
                .print();
        env.execute();
    }

    public static class MyTrigger extends Trigger<ClickEvent,TimeWindow>{

        //每条数据进入是触发一次
        @Override
        public TriggerResult onElement(ClickEvent element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            //设置标记位
            ValueState<Boolean> flag = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("flag", Types.BOOLEAN));
            if (flag.value() == null){
                long nextTs = element.ts  + 1000L -element.ts % 1000L;
                //注册一秒之后的定时器
                ctx.registerEventTimeTimer(nextTs);
                flag.update(true);

            }
            return TriggerResult.CONTINUE;
        }

        //当机器时间到达时定时器触发时
        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return null;
        }

        //当水位线时间到达时触发
        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            if (time < window.getEnd()){
                if (time + 1000L < window.getEnd()){
                    ctx.registerEventTimeTimer(time + 1000L);
                }
                return TriggerResult.FIRE;
            }
            return TriggerResult.CONTINUE;
        }

        //当窗口关闭时触发调用
        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
            ValueState<Boolean> flag = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("flag", Types.BOOLEAN));
            flag.clear();


        }
    }
}
